package edu.maxflow.mapreduce;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import edu.maxflow.Edge;
import edu.maxflow.Path;
import edu.maxflow.Vertex;
import edu.maxflow.mapreduce.util.MRConstants;

public class GraphMapper extends Mapper<LongWritable, Text, LongWritable, Vertex> {

	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		// get configuration variables
		Configuration conf = context.getConfiguration();
		long sourceVertexId = conf.getInt(MRConstants.SOURCE_VERTEX_ID, MRConstants.SOURCE_VERTEX_DEFAULT_VALUE);
		long targetVertexId = conf.getInt(MRConstants.TARGET_VERTEX_ID, MRConstants.TARGET_VERTEX_DEFAULT_VALUE);

		Vertex vertex = new Vertex();

		// read vertex info from the given text file in a pre-defined format
		String line = value.toString();
		String[] allEdgeInfo = line.split("\\s+");
		String[] neighbourVerticesInfo = allEdgeInfo[1].split(",");

		Integer vertexId = Integer.valueOf(allEdgeInfo[0].trim());
		vertex.setVertexId(vertexId);

		for (String neighbourVertexInfoString : neighbourVerticesInfo) {
			Edge edge = new Edge();

			String[] infoStringArray = neighbourVertexInfoString.split(":");
			edge.setCapacity(Integer.valueOf(infoStringArray[1].trim()));
			edge.setFlow(0);
			edge.setFromVertexId(vertexId);

			int toVertexId = Integer.valueOf(infoStringArray[0].trim());
			edge.setToVertexId(toVertexId);

			vertex.getAdjacentEdges().add(edge);

			if (toVertexId == sourceVertexId) {
				Path sourceExcessPath = new Path();
				sourceExcessPath.getEdgeList().add(edge);

				vertex.getSourceExcessPaths().add(sourceExcessPath);
			} else if (toVertexId == targetVertexId) {
				Path targetExcessPath = new Path();
				targetExcessPath.getEdgeList().add(edge);

				vertex.getTargetExcessPaths().add(targetExcessPath);
			}
		}

		context.write(new LongWritable(vertexId), vertex);
	}
}
